iT邦幫忙

2026 iThome 鐵人賽

DAY 11
1
AI Engineering

30天用 Claude Code + LangGraph 實作個人化 AI 學習教練系列 第 11 篇

Day 11:Thread 與 State 管理 - 讓對話有記憶

  • 分享至 

  • xImage
  •  

昨天的圖跑完一次就結束了,完全沒有記憶。你問它「我叫什麼名字」,它不會知道你上一句話有沒有自我介紹過,因為每次 graph.invoke(...) 都是全新的一次呼叫,State 用完就丟。

今天要解決這件事:讓使用者能跟教練進行多輪對話,而且就算把程式關掉、電腦重開機,之前聊過的內容還能接得回來。這對一個「陪你調整學習計畫」的教練來說是必要的,使用者不可能每次都要重新解釋一次自己的狀況。

為什麼需要 Thread

昨天的圖,你每呼叫一次 graph.invoke(...),State 就是全新的一份,跑完就消失,跟上一次呼叫完全沒有關聯。

但真實情況是:同一個使用者今天問一句、明天再問一句,教練要知道這是「同一段對話的延續」。而且系統以後會有很多使用者,甲的對話不能混進乙的對話裡。

Thread 就是用來解決這件事:每一段獨立的對話給一個 Thread ID,LangGraph 用這個 ID 來區分「這是誰的對話、存到哪裡、下次要接回哪一份」。

想像成醫院的病歷夾:每個病人有自己的病歷號碼(Thread ID),每次回診,醫生報上病歷號碼,就能調出這個人完整的看診紀錄,不會跟別的病人搞混。

Checkpoint 機制:LangGraph 怎麼自動存檔

Checkpoint(檢查點)是 LangGraph 內建的自動存檔功能。只要在編譯圖的時候給它一個「checkpointer」,圖每跑完一步,就會自動把當下的 State 存起來,存的時候會標記是哪個 Thread ID 的。

下次你用同一個 Thread ID 呼叫這張圖,LangGraph 會先去讀回之前存的 State,接著才把新的輸入疊上去,而不是從空白開始。這整個過程你完全不用自己寫存檔、讀檔的邏輯,只要在編譯時指定 checkpointer 就好。

第一次呼叫(Thread ID: user-1)
  ↓
State 是空的 → 加入這次的訊息 → 跑完 → 自動存檔

第二次呼叫(Thread ID: user-1,同一個 ID)
  ↓
先讀回之前存的 State → 加入這次的新訊息 → 跑完 → 再存一次

checkpointer 可以存在記憶體裡(程式關掉就消失,適合開發時快速測試),也可以存進 SQLite 檔案(存在硬碟上,重開機也還在)。今天直接用 SQLite,一步到位。

State 要改一下:從「一問一答」變成「訊息清單」

Day 10 的 State 只有 user_input 和 ai_response 兩個欄位,一次只能裝一輪對話。多輪對話需要把每一輪都留下來,所以今天要把 State 換成一個「訊息清單」,每輪對話結束後用「新增」的方式加進清單,而不是覆蓋掉。


實作步驟

步驟1:安裝 SQLite 版的 Checkpoint 套件

LangGraph 本身只內建「存在記憶體裡」的 checkpointer,要存進 SQLite 檔案需要另外裝一個套件:

cd backend
python -m pip install langgraph-checkpoint-sqlite

驗證:

python -c "from langgraph.checkpoint.sqlite import SqliteSaver; print('安裝成功')"

步驟2:改寫 State,改成訊息清單

檔案位置: backend/graph.py
狀態: 修改檔案(整個改寫,取代 Day 10 的內容)
用途: 定義能累積多輪對話的 StateGraph,並加上 SQLite Checkpoint
依賴: langgraph, langgraph-checkpoint-sqlite, langchain-ollama

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages


class ChatState(TypedDict):
    """messages 是一個清單,每輪對話會被「加進去」而不是覆蓋掉"""
    messages: Annotated[list, add_messages]

Annotated[list, add_messages] 是關鍵:add_messages 是 LangGraph 提供的規則,告訴它「這個欄位每次更新時,是把新訊息接在後面,不是整份換掉」。沒有這個標記的話,每次 Node 回傳新訊息都會把舊的洗掉,記憶就沒了。

步驟3:Node 邏輯幾乎不用改

檔案位置: backend/graph.py
狀態: 修改檔案(接續步驟2,繼續往下加)
用途: 定義呼叫 Ollama 的 Node,讓它一次看到完整對話歷史
依賴: langchain-ollama

from langchain_ollama import ChatOllama

_llm = ChatOllama(model="llama3.1:8b", temperature=0.3)


def call_model(state: ChatState) -> dict:
    """把目前為止所有訊息一起丟給模型,讓它看得到完整對話歷史"""
    response = _llm.invoke(state["messages"])
    return {"messages": [response]}

跟 Day 10 的差別:這次丟給模型的是 state["messages"](一整串歷史訊息),不是單一一句話,模型才能「記得」前面聊過什麼。

步驟4:加上 SQLite Checkpoint,組建圖

檔案位置: backend/graph.py
狀態: 修改檔案(接續步驟3,繼續往下加)
用途: 加上 SQLite Checkpoint,組建並編譯出最終的 graph
依賴: langgraph, langgraph-checkpoint-sqlite

import sqlite3
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver

# check_same_thread=False:允許這個連線被不同的呼叫共用,Checkpoint 內部需要這個設定
_conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
_checkpointer = SqliteSaver(_conn)

graph_builder = StateGraph(ChatState)
graph_builder.add_node("chat", call_model)
graph_builder.add_edge(START, "chat")
graph_builder.add_edge("chat", END)

# 編譯時把 checkpointer 傳進去,圖就會自動存檔
graph = graph_builder.compile(checkpointer=_checkpointer)

checkpoints.sqlite 是一個新的檔案,跟 Day 5 的 app.db 是不同的資料庫,一個專門存教練的使用者資料,一個專門存 LangGraph 的對話記憶,兩者互不相干。

完整的 graph.py 到這裡長這樣:

import sqlite3
from typing import Annotated
from typing_extensions import TypedDict
from langchain_ollama import ChatOllama
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver


class ChatState(TypedDict):
    """messages 是一個清單,每輪對話會被「加進去」而不是覆蓋掉"""
    messages: Annotated[list, add_messages]


_llm = ChatOllama(model="llama3.1:8b", temperature=0.3)


def call_model(state: ChatState) -> dict:
    """把目前為止所有訊息一起丟給模型,讓它看得到完整對話歷史"""
    response = _llm.invoke(state["messages"])
    return {"messages": [response]}


_conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
_checkpointer = SqliteSaver(_conn)

graph_builder = StateGraph(ChatState)
graph_builder.add_node("chat", call_model)
graph_builder.add_edge(START, "chat")
graph_builder.add_edge("chat", END)
graph = graph_builder.compile(checkpointer=_checkpointer)

步驟5:寫一支能重複執行的對話腳本

這支腳本每次執行都是「獨立的一個 Python 程式」,模擬使用者關掉終端機、下次再打開的情境。用固定的 Thread ID,證明對話歷史真的有被存下來。

檔案位置: backend/chat_once.py
狀態: 新增檔案
用途: 每次執行送一句話進去,印出目前為止的完整對話歷史
依賴: graph

import sys
from graph import graph


def main() -> None:
    """從命令列拿一句話,用固定的 Thread ID 送進圖裡"""
    if len(sys.argv) < 2:
        print('用法:python chat_once.py "你要說的話"')
        return

    user_message = sys.argv[1]
    config = {"configurable": {"thread_id": "demo-user"}}

    result = graph.invoke(
        {"messages": [{"role": "user", "content": user_message}]}, config
    )

    print("目前完整對話:")
    for msg in result["messages"]:
        print(f"  [{msg.type}] {msg.content}")


if __name__ == "__main__":
    main()

步驟6:測試多輪對話

先跑第一句:

python chat_once.py "我叫Alice,正在準備AWS Solutions Architect認證"

應該看到一問一答:

目前完整對話:
  [human] 我叫Alice,正在準備AWS Solutions Architect認證
  [ai] 你好Alice!準備AWS SA認證是個很棒的目標...

重點在這裡:完全關掉這個終端機視窗,開一個全新的終端機視窗,進入同一個 backend/ 目錄,跑第二句話:

python chat_once.py "我剛剛說我叫什麼名字?"

應該看到完整的四則訊息,而且模型的回答會正確答出「Alice」:

目前完整對話:
  [human] 我叫Alice,正在準備AWS Solutions Architect認證
  [ai] 你好Alice!準備AWS SA認證是個很棒的目標...
  [human] 我剛剛說我叫什麼名字?
  [ai] 你說你叫Alice。

這證明了兩件事:對話歷史真的存進了 checkpoints.sqlite,不是只存在記憶體裡;就算是完全獨立的兩次程式執行(等同於關掉重開),只要 Thread ID 一樣,歷史就接得回來。


常見問題

第二次執行時模型答不出正確的名字

先確認兩次執行用的是同一個 thread_id(範例裡固定寫死 "demo-user"),還有兩次都在同一個 backend/ 目錄下執行,確保用的是同一個 checkpoints.sqlite 檔案。

sqlite3.OperationalError: database is locked

通常是同時有兩個程式在寫同一個 checkpoints.sqlite。確認前一個 python chat_once.py 已經執行完畢、終端機已經回到提示字元,再執行下一次。

checkpoints.sqlite 檔案在哪裡?

在 backend/ 目錄下,第一次執行 chat_once.py 時會自動建立,可以用檔案總管或 ls 確認它存在。

ModuleNotFoundError: No module named 'langgraph.checkpoint.sqlite'

確認用 python -m pip install langgraph-checkpoint-sqlite 這個獨立套件,光裝 langgraph 本身是不夠的,SQLite 的 checkpoint 功能是分開打包的。

Annotated[list, add_messages] 這個寫法看不懂

把它想成一個特殊標記:告訴 LangGraph「這個欄位是清單,而且更新方式是『加在後面』」。不加這個標記的話,messages 欄位每次會被整個覆蓋掉,對話就會斷片,只記得最後一輪。


進度回顧

今天讓教練真正有了記憶。改寫了 State 讓它能累積多輪對話,加上 SQLite Checkpoint 讓對話存進硬碟,用兩次完全獨立的程式執行證明了記憶真的能跨越「關掉重開」這件事。

系統現在是這樣的:

Day 1 ✓ 產品定義完成
Day 2 ✓ 開發環境準備
Day 3 ✓ 專案架構設計
Day 4 ✓ 資料庫設計
Day 5 ✓ SQLite 資料庫建置
Day 6 ✓ FastAPI 基礎
Day 7 ✓ 使用者檔案 API
Day 8 ✓ 理解 LLM Agent 的本質
Day 9 ✓ 連接 Ollama 本機模型
Day 10 ✓ LangGraph 最小範例
Day 11 ✓ Thread 與 State 管理(今天)
Day 12 ⬜ 簡化的意圖路由

有了記憶之後,教練才有辦法真的「認得」使用者。明天(Day 12)要教它聽懂使用者在說什麼:這句話是要建立新計畫、回報進度,還是單純問問題,讓圖依照不同意圖走不同路線,這也是圖上第一次出現「分岔路」。


上一篇
Day 10:LangGraph 最小範例 - 組出第一個 Agent
下一篇
Day 12:簡化的意圖路由 - 教練的「聽力」
系列文
30天用 Claude Code + LangGraph 實作個人化 AI 學習教練 共 16 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

1 則留言

0
tsengyulun
iT邦新手 5 級 ‧ 2026-09-28 17:28:35

真的看到有下一頁了 超好笑

pst iT邦新手 5 級 ‧ 2026-09-28 20:05:51 檢舉

笨

我要留言

立即登入留言